package org.example.operator.action

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 *  案例实操
 *
 */
object Spark04_RDD_Operator_WordCount {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val context = new SparkContext(conf)

    val wordRdd: RDD[String] = context.makeRDD(
      List("Hello Spark", "Hello Scala"), 2)

    /**
      * 一些wordcount的练习，对之前练习过的转换算子和行动算子进行练习
      */

    def wordCount1(rdd:RDD[String]):Unit = {
    //  扁平化处理，将多个单词放在一起
      val wcRdd: RDD[String] = rdd.flatMap(_.split(""))
      // 将他们分组
      val unit: RDD[(String, Iterable[String])] = wcRdd.groupBy(v=>v)
      val result: RDD[(String, Int)] = unit.mapValues(v=>v.size)
    }

    def wordCount2(rdd:RDD[String]):Unit = {
      //  扁平化处理，将多个单词放在一起
      val wcRdd: RDD[String] = rdd.flatMap(_.split(""))

      val unit1: RDD[(String, Int)] = wcRdd.map(v=>(v,1))
      // 将他们分组
      val unit: RDD[(String, Iterable[Int])] = unit1.groupByKey()

      val result: RDD[(String, Int)] = unit.mapValues(v=>v.size)

    }

    def wordCount3(rdd:RDD[String]):Unit = {
      //  扁平化处理，将多个单词放在一起
      val wcRdd: RDD[String] = rdd.flatMap(_.split(""))

      val unit1: RDD[(String, Int)] = wcRdd.map(v=>(v,1))
      // 将他们分组
      val result: RDD[(String, Int)] = unit1.reduceByKey(_+_)
    }

    def wordCount4(rdd:RDD[String]):Unit = {
      //  扁平化处理，将多个单词放在一起
      val wcRdd: RDD[String] = rdd.flatMap(_.split(""))

      val unit1: RDD[(String, Int)] = wcRdd.map(v=>(v,1))
      // 将他们分组
      val result: RDD[(String, Int)] = unit1.aggregateByKey(0)(_+_,_+_)
    }

    def wordCount5(rdd:RDD[String]):Unit = {
      //  扁平化处理，将多个单词放在一起
      val wcRdd: RDD[String] = rdd.flatMap(_.split(""))

      val unit1: RDD[(String, Int)] = wcRdd.map(v=>(v,1))
      // 将他们分组
      val result: RDD[(String, Int)] = unit1.foldByKey(0)(_+_)
    }

    def wordCount6(rdd:RDD[String]):Unit = {
      //  扁平化处理，将多个单词放在一起
      val wcRdd: RDD[String] = rdd.flatMap(_.split(""))

      val unit1: RDD[(String, Int)] = wcRdd.map(v=>(v,1))
      // 将他们分组
      val result: RDD[(String, Int)] = unit1.combineByKey(
        v=>v,
        (x,y)=>x+y,
        (x,y)=>x+y,
      )
    }

    def wordCount7(rdd:RDD[String]):Unit = {
      //  扁平化处理，将多个单词放在一起
      val wcRdd: RDD[String] = rdd.flatMap(_.split(""))

      val unit1: RDD[(String, Int)] = wcRdd.map(v=>(v,1))
      // 将他们分组
      val stringToLong: collection.Map[String, Long] = unit1.countByKey()
    }

    def wordCount8(rdd:RDD[String]):Unit = {
      //  扁平化处理，将多个单词放在一起
      val wcRdd: RDD[String] = rdd.flatMap(_.split(""))
      val stringToLong: collection.Map[String, Long] = wcRdd.countByValue()
    }

    def wordCount9(rdd:RDD[String]):Unit = {
      //  扁平化处理，将多个单词放在一起
      val wcRdd: RDD[String] = rdd.flatMap(_.split(""))

      val unit: RDD[mutable.Map[String, Long]] = wcRdd.map(word => {
        mutable.Map[String, Long]((word, 1))
      })
      unit.reduce(
        (map1,map2)=>{
           map2.foreach{
             case (word,count)=>{
               val c:Long = map1.getOrElse(word,0L)+count
               map1.update(word,c)
             }
           }
            map1
        }
      )
    }



    context.stop()
  }
}
